package com.bins.reactor.demo1;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Exceptions;

class MySubscriber implements Subscriber<String> {

    /**
     * 订阅管理
     */
    volatile Subscription subscription;

    /**
     * 每次请求条数
     */
    final int i = 2;

    /**
     * 一次请求中的 出版商当前是第几次给订阅者发送消息
     */
    int j = 0;

    @Override
    public void onSubscribe(Subscription subscription) {
        System.out.println("已发起订阅");
        this.subscription = subscription;
        subscription.request(i);

    }

    @Override
    public void onNext(String value) {
        j = j + 1;
        System.out.println("出版商发送来了数据====" + value + "===第" + j + "条");
        if (j == i) {
            j = 0;
            subscription.request(i);
        }
    }

    @Override
    public void onComplete() {
        System.out.println("数据已发送完");
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("数据发送异常");
        throw Exceptions.errorCallbackNotImplemented(throwable);
    }

}
